package com.geekstudy.week13;


import com.geekstudy.week13.kafka.KafkaConsumerService;
import com.geekstudy.week13.kafka.KafkaProducerService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.kafka.core.KafkaTemplate;

import java.time.Duration;

@SpringBootApplication
public class Week13SpringBootMain {

    static final Logger logger = LoggerFactory.getLogger(Week13SpringBootMain.class);

    public static void main(String[] args) {

        ConfigurableApplicationContext configurableApplicationContext = SpringApplication.run(Week13SpringBootMain.class, args);
        logger.info(" beanCount:{} ", configurableApplicationContext.getBeanDefinitionCount());

        String topic = "topic.demo1";
        KafkaProducerService kafkaProducerService = configurableApplicationContext.getBean(KafkaProducerService.class);
        String message = "Demo send 202109";
        kafkaProducerService.send(message, topic);

        KafkaConsumerService kafkaConsumerService = configurableApplicationContext.getBean(KafkaConsumerService.class);

        KafkaConsumer<String, String> kafkaConsumer = configurableApplicationContext.getBean(KafkaConsumer.class);
        ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
        if(consumerRecords != null && consumerRecords.count() > 0) {
            for(ConsumerRecord<String, String> consumerRecord : consumerRecords.records(topic)) {
                kafkaConsumerService.topicTest(consumerRecord, null, topic);
            }
        }







    }
}
